package org.rcy.framework.mq.config.rocket;

import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.MQAdmin;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.spring.autoconfigure.ExtProducerResetConfiguration;
import org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.autoconfigure.RocketMQTransactionConfiguration;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.rcy.framework.mq.converter.MessageConverterConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.core.env.Environment;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;

/**
 * @author 吴康桥
 * @Description
 * @date 2022/11/11 17:47
 */

@Configuration
@EnableConfigurationProperties({RocketMQProperties.class})
@ConditionalOnClass({MQAdmin.class})
@ConditionalOnProperty(
		prefix = "rocketmq",
		value = {"name-server"},
		matchIfMissing = false
)
@Import({MessageConverterConfiguration.class, ListenerContainerConfiguration.class, ExtProducerResetConfiguration.class, RocketMQTransactionConfiguration.class})
@AutoConfigureAfter({MessageConverterConfiguration.class})
@AutoConfigureBefore({RocketMQTransactionConfiguration.class})
public class SpringRocketAutoConfiguration {
	private static final Logger log = LoggerFactory.getLogger(SpringRocketAutoConfiguration.class);
	public static final String ROCKETMQ_TEMPLATE_DEFAULT_GLOBAL_NAME = "rocketMQTemplate";
	@Autowired
	private Environment environment;

	public SpringRocketAutoConfiguration() {
	}

	@PostConstruct
	public void checkProperties() {
		String nameServer = (String)this.environment.getProperty("rocketmq.name-server", String.class);
		log.debug("rocketmq.nameServer = {}", nameServer);
		if (nameServer == null) {
			log.warn("The necessary spring property 'rocketmq.name-server' is not defined, all rockertmq beans creation are skipped!");
		}

	}

	@Bean
	@ConditionalOnMissingBean({DefaultMQProducer.class})
	@ConditionalOnProperty(
			prefix = "rocketmq",
			value = {"name-server", "producer.group"}
	)
	public DefaultMQProducer defaultMQProducer(RocketMQProperties rocketMQProperties) {
		RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer();
		String nameServer = rocketMQProperties.getNameServer();
		String groupName = producerConfig.getGroup();
		Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
		Assert.hasText(groupName, "[rocketmq.producer.group] must not be null");
		String accessChannel = rocketMQProperties.getAccessChannel();
		String ak = rocketMQProperties.getProducer().getAccessKey();
		String sk = rocketMQProperties.getProducer().getSecretKey();
		boolean isEnableMsgTrace = rocketMQProperties.getProducer().isEnableMsgTrace();
		String customizedTraceTopic = rocketMQProperties.getProducer().getCustomizedTraceTopic();
		DefaultMQProducer producer = RocketMQUtil.createDefaultMQProducer(groupName, ak, sk, isEnableMsgTrace, customizedTraceTopic);
		producer.setNamesrvAddr(nameServer);
		if (!StringUtils.isEmpty(accessChannel)) {
			producer.setAccessChannel(AccessChannel.valueOf(accessChannel));
		}

		producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
		producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
		producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
		producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
		producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
		producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
		return producer;
	}

	@Bean(
			destroyMethod = "destroy"
	)
	@ConditionalOnBean({DefaultMQProducer.class})
	@ConditionalOnMissingBean(
			name = {"rocketMQTemplate"}
	)
	public RocketMQTemplate rocketMQTemplate(DefaultMQProducer mqProducer, RocketMQMessageConverter rocketMQMessageConverter) {
		RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
		rocketMQTemplate.setProducer(mqProducer);
		rocketMQTemplate.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
		return rocketMQTemplate;
	}
}
